package com.uuah.server.transport;

import java.beans.ExceptionListener;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import javax.resource.spi.work.WorkEvent;
import javax.resource.spi.work.WorkListener;

import org.apache.commons.pool.KeyedPoolableObjectFactory;
import org.apache.commons.pool.impl.GenericKeyedObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.uuah.api.config.IThreadingProfile;
import com.uuah.api.work.IWorkManager;
import com.uuah.concurrent.NamedThreadFactory;
import com.uuah.concurrent.WaitableBoolean;
import com.uuah.config.i18n.CoreMessages;
import com.uuah.config.i18n.MessageFactory;
import com.uuah.exception.AbstractUuahException;
import com.uuah.exception.UuahRuntimeException;
import com.uuah.exception.lifecycle.DisposeException;
import com.uuah.exception.lifecycle.InitialisationException;
import com.uuah.exception.lifecycle.LifecycleException;
import com.uuah.server.UuahServer;
import com.uuah.server.api.IUuahContext;
import com.uuah.server.api.IUuahEvent;
import com.uuah.server.api.IUuahMessage;
import com.uuah.server.api.component.service.IService;
import com.uuah.server.api.context.notification.IServerNotificationHandler;
import com.uuah.server.api.endpoint.IEndpointURI;
import com.uuah.server.api.endpoint.IUuahEndpoint;
import com.uuah.server.api.session.ISessionHandler;
import com.uuah.server.api.transformer.IBaseTransformer;
import com.uuah.server.api.transport.IConnectable;
import com.uuah.server.api.transport.IConnector;
import com.uuah.server.api.transport.IMessageAdapter;
import com.uuah.server.api.transport.IMessageDispatcher;
import com.uuah.server.api.transport.IMessageDispatcherFactory;
import com.uuah.server.api.transport.IMessageReceiver;
import com.uuah.server.api.transport.retry.IRetryCallback;
import com.uuah.server.api.transport.retry.IRetryContext;
import com.uuah.server.api.transport.retry.RetryPolicyTemplate;
import com.uuah.server.api.transport.service.ITransportServiceDescriptor;
import com.uuah.server.config.Constants;
import com.uuah.server.config.i18n.ServerMessages;
import com.uuah.server.config.registry.ServiceDescriptorFactory;
import com.uuah.server.context.notification.ConnectionNotification;
import com.uuah.server.context.notification.ServerNotification;
import com.uuah.server.exception.AbstractUuahServerException;
import com.uuah.server.exception.ExceptionStrategy;
import com.uuah.server.exception.MessagingException;
import com.uuah.server.exception.config.ConfigurationException;
import com.uuah.server.exception.registry.ServiceException;
import com.uuah.server.exception.transport.ConnectorException;
import com.uuah.server.exception.transport.DispatchException;
import com.uuah.server.exception.transport.service.TransportServiceException;
import com.uuah.server.session.UuahSessionHandler;
import com.uuah.server.transformer.TransformerUtils;
import com.uuah.server.util.UuahServerUtils;
import com.uuah.utils.ClassUtils;
import com.uuah.utils.ObjectUtils;
import com.uuah.utils.StringUtils;

/**
 *
 * TODO comment ...
 *
 * <p>
 * update record:updator，update time，update content and version
 * </p>
 *
 * @author <a href="jonny_quan@hotmail.com">jonny</a>
 * @date 2009 2009-3-27 下午02:49:23
 * @version 1.0.0
 */
public abstract class AbstractConnector implements IConnector,
		ExceptionListener, IConnectable, WorkListener {

	protected transient final Logger logger = LoggerFactory
			.getLogger(getClass());
	protected final AtomicBoolean started = new AtomicBoolean(false);

	protected final AtomicBoolean initialised = new AtomicBoolean(false);

	protected volatile String name;

	protected volatile ExceptionListener exceptionListener;

	protected final AtomicBoolean disposed = new AtomicBoolean(false);

	protected final AtomicBoolean disposing = new AtomicBoolean(false);

	protected volatile IMessageDispatcherFactory dispatcherFactory;

	protected final GenericKeyedObjectPool dispatchers = new GenericKeyedObjectPool();

	protected final ConcurrentMap receivers = new ConcurrentHashMap();

	private volatile IThreadingProfile dispatcherThreadingProfile;

	private volatile IThreadingProfile receiverThreadingProfile;

	private RetryPolicyTemplate retryPolicyTemplate;

	protected final Semaphore connectedSemaphore = new Semaphore(0);

	protected final WaitableBoolean connected = new WaitableBoolean(false);

	protected final WaitableBoolean connecting = new WaitableBoolean(false);

	protected final WaitableBoolean startOnConnect = new WaitableBoolean(false);

	private IServerNotificationHandler notificationHandler;

	private final List supportedProtocols;

	private final AtomicReference receiverWorkManager = new AtomicReference();

	private final AtomicReference dispatcherWorkManager = new AtomicReference();

	private final AtomicReference scheduler = new AtomicReference();

	protected volatile ITransportServiceDescriptor serviceDescriptor;

	protected volatile ISessionHandler sessionHandler = new UuahSessionHandler();

	protected IUuahContext uuahContext;

	public AbstractConnector() {
		supportedProtocols = new ArrayList();
		supportedProtocols.add(getProtocol().toLowerCase());

		dispatchers.setTestOnBorrow(false);
		dispatchers.setTestOnReturn(true);
	}

	public String getName() {
		return name;
	}

	public void setName(String newName) {
		if (newName == null) {
			throw new IllegalArgumentException(CoreMessages.objectIsNull(
					"Connector name").toString());
		}

		if (logger.isDebugEnabled()) {
			logger.debug("Set Connector name to: " + newName);
		}

		name = newName;
	}

	public final synchronized void initialise() throws InitialisationException {
		if (initialised.get()) {
			InitialisationException e = new InitialisationException(
					ServerMessages.objectAlreadyInitialised("Connector '"
							+ getProtocol() + "." + getName() + "'"), this);
			throw e;
		}

		if (logger.isInfoEnabled()) {
			logger.info("Initialising: " + this);
		}

		if (retryPolicyTemplate == null) {
			retryPolicyTemplate = (RetryPolicyTemplate) uuahContext
					.getRegistry().lookupObject(
							Constants.OBJECT_DEFAULT_RETRY_POLICY_TEMPLATE);
		}

		this.initFromServiceDescriptor();

		setMaxDispatchersActive(getDispatcherThreadingProfile()
				.getMaxThreadsActive());

		this.doInitialise();

		if (exceptionListener == null) {
			exceptionListener = new ExceptionStrategy();
			((ExceptionStrategy) exceptionListener).setUuahContext(uuahContext);
			((ExceptionStrategy) exceptionListener).initialise();
		}

		try {
			initWorkManagers();
		} catch (AbstractUuahException e) {
			throw new InitialisationException(e, this);
		}
		initialised.set(true);
	}

	public final synchronized void start() throws AbstractUuahException {
		this.checkDisposed();

		if (!this.isConnected()) {
			startOnConnect.set(true);

			try {
				connect();
			} catch (Exception e) {
				throw new LifecycleException(e, this);
			}
			return;
		}

		if (logger.isInfoEnabled()) {
			logger.info("Starting: " + this);
		}

		ScheduledExecutorService currentScheduler = (ScheduledExecutorService) scheduler
				.get();
		if (currentScheduler == null || currentScheduler.isShutdown()) {
			scheduler.set(this.getScheduler());
		}

		this.doStart();
		started.set(true);

		if (receivers != null) {
			for (Iterator iterator = receivers.values().iterator(); iterator
					.hasNext();) {
				IMessageReceiver mr = (IMessageReceiver) iterator.next();
				if (logger.isDebugEnabled()) {
					logger.debug("Starting receiver on endpoint: "
							+ mr.getEndpoint().getEndpointURI());
				}
				mr.start();
			}
		}

		if (logger.isInfoEnabled()) {
			logger.info("Started: " + this);
		}
	}

	public final boolean isStarted() {
		return started.get();
	}

	public final synchronized void stop() throws AbstractUuahException {
		if (this.isDisposed()) {
			return;
		}

		if (this.isStarted()) {
			if (logger.isInfoEnabled()) {
				logger.info("Stopping: " + this);
			}

			// shutdown our scheduler service
			((ScheduledExecutorService) scheduler.get()).shutdown();

			this.doStop();
			started.set(false);

			// Stop all the receivers on this connector (this will cause them to
			// disconnect too)
			if (receivers != null) {
				for (Iterator iterator = receivers.values().iterator(); iterator
						.hasNext();) {
					IMessageReceiver mr = (IMessageReceiver) iterator.next();
					if (logger.isDebugEnabled()) {
						logger.debug("Stopping receiver on endpoint: "
								+ mr.getEndpoint().getEndpointURI());
					}
					mr.stop();
				}
			}
		}

		if (this.isConnected()) {
			try {
				this.disconnect();
			} catch (Exception e) {
				logger.error("Failed to disconnect: " + e.getMessage(), e);
			}
		}

		// make sure the scheduler is gone
		scheduler.set(null);

		if (logger.isInfoEnabled()) {
			logger.info("Stopped: " + this);
		}
	}

	public final synchronized void dispose() {
		disposing.set(true);

		if (logger.isInfoEnabled()) {
			logger.info("Disposing: " + this);
		}

		try {
			this.stop();
		} catch (AbstractUuahException e) {
			logger.warn("Failed to stop during shutdown: " + e.getMessage(), e);
		}

		this.disposeReceivers();
		this.disposeDispatchers();
		this.disposeWorkManagers();

		this.doDispose();
		disposed.set(true);
		initialised.set(false);

		if (logger.isInfoEnabled()) {
			logger.info("Disposed: " + this);
		}
	}

	protected void initWorkManagers() throws AbstractUuahException {
		if (receiverWorkManager.get() == null) {
			IThreadingProfile tp = getReceiverThreadingProfile();
			if (tp == null) {
				throw new ConfigurationException(
						MessageFactory
								.createStaticMessage("No ReceiverThreadingProfile has been configured"));
			}
			IWorkManager newWorkManager = tp.createWorkManager(getName()
					+ ".receiver");

			if (receiverWorkManager.compareAndSet(null, newWorkManager)) {
				newWorkManager.start();
			}
		}

		if (dispatcherWorkManager.get() == null) {
			IWorkManager newWorkManager = this.getDispatcherThreadingProfile()
					.createWorkManager(getName() + ".dispatcher");

			if (dispatcherWorkManager.compareAndSet(null, newWorkManager)) {
				newWorkManager.start();
			}
		}
	}

	protected void disposeWorkManagers() {
		logger.debug("Disposing dispatcher work manager");
		IWorkManager workManager = (IWorkManager) dispatcherWorkManager.get();
		if (workManager != null) {
			workManager.dispose();
		}
		dispatcherWorkManager.set(null);

		logger.debug("Disposing receiver work manager");
		workManager = (IWorkManager) receiverWorkManager.get();
		if (workManager != null) {
			workManager.dispose();
		}
		receiverWorkManager.set(null);
	}

	protected void disposeReceivers() {
		if (receivers != null) {
			logger.debug("Disposing Receivers");

			for (Iterator iterator = receivers.values().iterator(); iterator
					.hasNext();) {
				IMessageReceiver receiver = (IMessageReceiver) iterator.next();

				try {
					this.destroyReceiver(receiver, receiver.getEndpoint());
				} catch (Throwable e) {
					logger.error("Failed to destroy receiver: " + receiver, e);
				}
			}

			receivers.clear();
			logger.debug("Receivers Disposed");
		}
	}

	protected void disposeDispatchers() {
		if (dispatchers != null) {
			logger.debug("Disposing Dispatchers");
			dispatchers.clear();
			logger.debug("Dispatchers Disposed");
		}
	}

	public boolean isDisposed() {
		return disposed.get();
	}

	public void handleException(Exception exception) {
		if (exceptionListener == null) {
			throw new UuahRuntimeException(ServerMessages
					.exceptionOnConnectorNotExceptionListener(this.getName()),
					exception);
		} else {
			exceptionListener.exceptionThrown(exception);
		}
	}

	public void exceptionThrown(Exception e) {
		handleException(e);
	}

	public ExceptionListener getExceptionListener() {
		return exceptionListener;
	}

	public void setExceptionListener(ExceptionListener listener) {
		exceptionListener = listener;
	}

	public IMessageDispatcherFactory getDispatcherFactory() {
		return dispatcherFactory;
	}

	public void setDispatcherFactory(IMessageDispatcherFactory dispatcherFactory) {
		KeyedPoolableObjectFactory poolFactory;

		if (dispatcherFactory instanceof KeyedPoolableObjectFactory) {
			poolFactory = (KeyedPoolableObjectFactory) dispatcherFactory;
		} else {
			poolFactory = new KeyedPoolMessageDispatcherFactoryAdapter(
					dispatcherFactory);
		}

		this.dispatchers.setFactory(poolFactory);

		this.dispatcherFactory = dispatcherFactory;
	}

	public int getMaxDispatchersActive() {
		return this.dispatchers.getMaxActive();
	}

	public void setMaxDispatchersActive(int maxActive) {
		this.dispatchers.setMaxActive(maxActive);
		this.dispatchers.setMaxIdle(maxActive);
	}

	private IMessageDispatcher getDispatcher(IUuahEndpoint endpoint)
			throws AbstractUuahServerException {
		try {
			this.checkDisposed();
		} catch (DisposeException e) {
			throw new AbstractUuahServerException(e);
		}

		if (endpoint == null) {
			throw new IllegalArgumentException("Endpoint must not be null");
		}

		if (!supportsProtocol(endpoint.getConnector().getProtocol())) {
			throw new IllegalArgumentException(ServerMessages
					.connectorSchemeIncompatibleWithEndpointScheme(
							this.getProtocol(),
							endpoint.getEndpointURI().toString()).getMessage());
		}

		IMessageDispatcher dispatcher = null;
		try {
			if (logger.isDebugEnabled()) {
				logger.debug("Borrowing a dispatcher for endpoint: "
						+ endpoint.getEndpointURI());
			}

			dispatcher = (IMessageDispatcher) dispatchers
					.borrowObject(endpoint);
			dispatcher.initialise();

			if (logger.isDebugEnabled()) {
				logger.debug("Borrowed a dispatcher for endpoint: "
						+ endpoint.getEndpointURI() + " = "
						+ dispatcher.toString());
			}

			return dispatcher;
		} catch (Exception ex) {
			throw new ConnectorException(ServerMessages.connectorCausedError(),
					this, ex);
		} finally {
			try {
				if (logger.isDebugEnabled()) {
					logger.debug("Borrowed dispatcher: "
							+ ObjectUtils.toString(dispatcher, "null"));
				}
			} catch (Exception ex) {
				throw new ConnectorException(ServerMessages
						.connectorCausedError(), this, ex);
			}
		}
	}

	private void returnDispatcher(IUuahEndpoint endpoint,
			IMessageDispatcher dispatcher) {
		if (endpoint != null && dispatcher != null) {
			try {
				if (logger.isDebugEnabled()) {
					logger.debug("Returning dispatcher for endpoint: "
							+ endpoint.getEndpointURI() + " = "
							+ dispatcher.toString());
				}

			} catch (Exception ex) {
				// Logging failed
			} finally {
				try {
					dispatchers.returnObject(endpoint, dispatcher);
				} catch (Exception e) {
					logger
							.error(
									"Failed to dispose dispatcher for endpoint: "
											+ endpoint
											+ ". This will cause a memory leak. Please report to",
									e);
				}
			}
		}
	}

	protected void checkDisposed() throws DisposeException {
		if (this.isDisposed()) {
			throw new DisposeException(ServerMessages
					.cannotUseDisposedConnector(), this);
		}
	}

	public IMessageReceiver registerListener(IService service,
			IUuahEndpoint endpoint) throws Exception {
		if (endpoint == null) {
			throw new IllegalArgumentException(
					"The endpoint cannot be null when registering a listener");
		}

		if (service == null) {
			throw new IllegalArgumentException(
					"The service cannot be null when registering a listener");
		}

		IEndpointURI endpointUri = endpoint.getEndpointURI();
		if (endpointUri == null) {
			throw new ConnectorException(ServerMessages
					.endpointIsNullForListener(), this);
		}

		logger.warn("Registering listener: " + service.getName()
				+ " on endpointUri: " + endpointUri.toString());

		if (getReceiver(service, endpoint) != null) {
			throw new ConnectorException(ServerMessages
					.listenerAlreadyRegistered(endpointUri), this);
		}

		IMessageReceiver receiver = createReceiver(service, endpoint);
		Object receiverKey = getReceiverKey(service, endpoint);
		receiver.setReceiverKey(receiverKey.toString());
		receiver.initialise();
		receivers.put(receiverKey, receiver);

		return receiver;
	}

	protected Object getReceiverKey(IService service, IUuahEndpoint endpoint) {
		return StringUtils.defaultIfEmpty(endpoint.getEndpointURI()
				.getFilterAddress(), endpoint.getEndpointURI().getAddress());
	}

	public final void unregisterListener(IService service, IUuahEndpoint endpoint)
			throws Exception {
		if (service == null) {
			throw new IllegalArgumentException(
					"The service must not be null when you unregister a listener");
		}

		if (endpoint == null) {
			throw new IllegalArgumentException(
					"The endpoint must not be null when you unregister a listener");
		}

		IEndpointURI endpointUri = endpoint.getEndpointURI();
		if (endpointUri == null) {
			throw new IllegalArgumentException(
					"The endpointUri must not be null when you unregister a listener");
		}

		if (logger.isInfoEnabled()) {
			logger.info("Removing listener on endpointUri: " + endpointUri);
		}

		if (receivers != null && !receivers.isEmpty()) {
			IMessageReceiver receiver = (IMessageReceiver) receivers
					.remove(getReceiverKey(service, endpoint));
			if (receiver != null) {
				destroyReceiver(receiver, endpoint);
				receiver.dispose();
			}
		}
	}

	public IThreadingProfile getDispatcherThreadingProfile() {
		if (dispatcherThreadingProfile == null && uuahContext != null) {
			dispatcherThreadingProfile = uuahContext
					.getDefaultMessageDispatcherThreadingProfile();
		}
		return dispatcherThreadingProfile;
	}

	public void setDispatcherThreadingProfile(
			IThreadingProfile dispatcherThreadingProfile) {
		this.dispatcherThreadingProfile = dispatcherThreadingProfile;
	}

	public IThreadingProfile getReceiverThreadingProfile() {
		if (receiverThreadingProfile == null && uuahContext != null) {
			receiverThreadingProfile = uuahContext
					.getDefaultMessageReceiverThreadingProfile();
		}

		return receiverThreadingProfile;
	}

	public void setReceiverThreadingProfile(
			IThreadingProfile receiverThreadingProfile) {
		this.receiverThreadingProfile = receiverThreadingProfile;
	}

	public void destroyReceiver(IMessageReceiver receiver, IUuahEndpoint endpoint)
			throws Exception {
		receiver.dispose();
	}

	protected abstract void doInitialise() throws InitialisationException;

	protected abstract void doDispose();

	protected abstract void doStart() throws AbstractUuahException;

	protected abstract void doStop() throws AbstractUuahException;

	public IBaseTransformer getDefaultRequestTransformer() {
		if (serviceDescriptor == null) {
			throw new RuntimeException("serviceDescriptor not initialized");
		}
		return TransformerUtils.getDefaultRequestTransformer(serviceDescriptor);
	}

	public IBaseTransformer getDefaultResponseTransformer() {
		if (serviceDescriptor == null) {
			throw new RuntimeException("serviceDescriptor not initialized");
		}
		return TransformerUtils
				.getDefaultResponseTransformer(serviceDescriptor);
	}

	public void fireNotification(ServerNotification notification) {
		notificationHandler.fireNotification(notification);
	}

	public boolean isDisposing() {
		return disposing.get();
	}

	public boolean isSyncEnabled(String protocol) {
		return false;
	}

	public IMessageReceiver getReceiver(IService service, IUuahEndpoint endpoint) {
		if (receivers != null) {
			Object key = getReceiverKey(service, endpoint);
			if (key != null) {
				return (IMessageReceiver) receivers.get(key);
			} else {
				throw new RuntimeException(
						"getReceiverKey() returned a null key");
			}
		} else {
			throw new RuntimeException("Connector has not been initialized.");
		}
	}

	public IMessageReceiver lookupReceiver(String key) {
		if (key != null) {
			return (IMessageReceiver) receivers.get(key);
		} else {
			throw new IllegalArgumentException("Receiver key must not be null");
		}
	}

	public void connect() throws Exception {
		this.checkDisposed();

		if (isConnected() || isConnecting()) {
			return;
		}

		retryPolicyTemplate.execute(new IRetryCallback() {
			public void doWork(IRetryContext context) throws Exception {
				setConnecting(true);
				doConnect();
				setConnected(true);
				setConnecting(false);
			}

			public String getWorkDescription() {
				return getConnectionDescription();
			}
		});

		// �ȴ����ӻ����Բ����̳߳����꣬�׳��쳣
		connected.whenTrue(null);

		if (startOnConnect.get()) {
			start();
		}

		if (receivers != null) {
			for (Iterator iterator = receivers.values().iterator(); iterator
					.hasNext();) {
				IMessageReceiver receiver = (IMessageReceiver) iterator.next();
				if (logger.isDebugEnabled()) {
					logger.debug("Connecting receiver on endpoint: "
							+ receiver.getEndpoint().getEndpointURI());
				}
				receiver.connect();
			}
		}
	}

	public void disconnect() throws Exception {
		startOnConnect.set(this.isStarted());

		this.fireNotification(new ConnectionNotification(this,
				getConnectEventId(),
				ConnectionNotification.CONNECTION_DISCONNECTED));

		connected.set(false);
		connectedSemaphore.drainPermits();

		try {
			this.doDisconnect();
		} finally {
			this.stop();
		}

		logger.info("Disconnected: " + this.getConnectionDescription());
	}

	public String getConnectionDescription() {
		return this.toString();
	}

	public final boolean isConnected() {
		return connected.get();
	}

	public final void setConnected(boolean flag) {
		connected.set(flag);
	}

	public final boolean isConnecting() {
		return connecting.get();
	}

	protected final void setConnecting(boolean flag) {
		connecting.set(flag);
	}

	protected abstract void doConnect() throws Exception;

	protected abstract void doDisconnect() throws Exception;

	protected String getConnectEventId() {
		return getName();
	}

	public void registerSupportedProtocol(String protocol) {
		protocol = protocol.toLowerCase();
		if (protocol.startsWith(getProtocol().toLowerCase())) {
			registerSupportedProtocolWithoutPrefix(protocol);
		} else {
			supportedProtocols
					.add(getProtocol().toLowerCase() + ":" + protocol);
		}
	}

	protected void registerSupportedProtocolWithoutPrefix(String protocol) {
		supportedProtocols.add(protocol.toLowerCase());
	}

	public void unregisterSupportedProtocol(String protocol) {
		protocol = protocol.toLowerCase();
		if (protocol.startsWith(getProtocol().toLowerCase())) {
			supportedProtocols.remove(protocol);
		} else {
			supportedProtocols.remove(getProtocol().toLowerCase() + ":"
					+ protocol);
		}
	}

	public boolean supportsProtocol(String protocol) {
		return supportedProtocols.contains(protocol.toLowerCase());
	}

	public List getSupportedProtocols() {
		return Collections.unmodifiableList(supportedProtocols);
	}

	public void setSupportedProtocols(List supportedProtocols) {
		for (Iterator iterator = supportedProtocols.iterator(); iterator
				.hasNext();) {
			String s = (String) iterator.next();
			registerSupportedProtocol(s);
		}
	}

	protected IWorkManager getReceiverWorkManager(String receiverName)
			throws AbstractUuahServerException {
		return (IWorkManager) receiverWorkManager.get();
	}

	protected IWorkManager getDispatcherWorkManager()
			throws AbstractUuahServerException {
		return (IWorkManager) dispatcherWorkManager.get();
	}

	public ScheduledExecutorService getScheduler() {
		if (scheduler.get() == null) {
			ThreadFactory threadFactory = new NamedThreadFactory(this.getName()
					+ ".scheduler");
			ScheduledThreadPoolExecutor newExecutor = new ScheduledThreadPoolExecutor(
					4, threadFactory);
			newExecutor
					.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
			newExecutor.setKeepAliveTime(this.getReceiverThreadingProfile()
					.getThreadTTL(), TimeUnit.MILLISECONDS);
			newExecutor.allowCoreThreadTimeOut(true);

			if (!scheduler.compareAndSet(null, newExecutor)) {
				newExecutor.shutdown();
			}
		}

		return (ScheduledExecutorService) scheduler.get();
	}

	public ISessionHandler getSessionHandler() {
		return sessionHandler;
	}

	public void setSessionHandler(ISessionHandler sessionHandler) {
		this.sessionHandler = sessionHandler;
	}

	public void workAccepted(WorkEvent event) {
		this.handleWorkException(event, "workAccepted");
	}

	public void workRejected(WorkEvent event) {
		this.handleWorkException(event, "workRejected");
	}

	public void workStarted(WorkEvent event) {
		this.handleWorkException(event, "workStarted");
	}

	public void workCompleted(WorkEvent event) {
		this.handleWorkException(event, "workCompleted");
	}

	protected void handleWorkException(WorkEvent event, String type) {
		if (event == null) {
			return;
		}

		Throwable e = event.getException();

		if (e == null) {
			return;
		}

		if (e.getCause() != null) {
			e = e.getCause();
		}

		logger.error("Work caused exception on '" + type
				+ "'. Work being executed was: " + event.getWork().toString());

		if (e instanceof Exception) {
			this.handleException((Exception) e);
		} else {
			throw new UuahRuntimeException(ServerMessages
					.connectorCausedError(this.getName()), e);
		}
	}

	public void dispatch(IUuahEndpoint endpoint, IUuahEvent event)
			throws DispatchException {
		IMessageDispatcher dispatcher = null;

		try {
			dispatcher = this.getDispatcher(endpoint);
			dispatcher.dispatch(event);
		} catch (DispatchException dex) {
			throw dex;
		} catch (AbstractUuahServerException ex) {
			throw new DispatchException(event.getMessage(), endpoint, ex);
		} finally {
			this.returnDispatcher(endpoint, dispatcher);
		}
	}

	public IUuahMessage send(IUuahEndpoint endpoint, IUuahEvent event)
			throws DispatchException {
		IMessageDispatcher dispatcher = null;

		try {
			dispatcher = this.getDispatcher(endpoint);
			return dispatcher.send(event);
		} catch (DispatchException dex) {
			throw dex;
		} catch (AbstractUuahException ex) {
			throw new DispatchException(event.getMessage(), endpoint, ex);
		} finally {
			this.returnDispatcher(endpoint, dispatcher);
		}
	}

	public void initialiseFromUrl(IEndpointURI endpointUri)
			throws InitialisationException {
		if (!supportsProtocol(endpointUri.getFullScheme())) {
			throw new InitialisationException(ServerMessages
					.schemeNotCompatibleWithConnector(endpointUri
							.getFullScheme(), this.getClass()), this);
		}
		Properties props = new Properties();
		props.putAll(endpointUri.getParams());

		if (endpointUri.getUserInfo() != null) {
			props.setProperty("username", endpointUri.getUser());
			String passwd = endpointUri.getPassword();
			if (passwd != null) {
				props.setProperty("password", passwd);
			}
		}
		String host = endpointUri.getHost();
		if (host != null) {
			props.setProperty("hostname", host);
			props.setProperty("host", host);
		}
		if (endpointUri.getPort() > -1) {
			props.setProperty("port", String.valueOf(endpointUri.getPort()));
		}

		com.uuah.utils.BeanUtils.populateWithoutFail(this, props, true);

		setName(UuahServerUtils.getConnectorName(this));
	}

	protected synchronized void initFromServiceDescriptor()
			throws InitialisationException {
		try {
			serviceDescriptor = (ITransportServiceDescriptor) UuahServer
					.getUuahContext().getRegistry().lookupServiceDescriptor(
							ServiceDescriptorFactory.PROVIDER_SERVICE_TYPE,
							getProtocol().toLowerCase());
			if (serviceDescriptor == null) {
				throw new ServiceException(ServerMessages
						.noServiceTransportDescriptor(getProtocol()));
			}

			if (logger.isDebugEnabled()) {
				logger.debug("Loading DispatcherFactory for connector: "
						+ getName() + " (" + getClass().getName() + ")");
			}

			IMessageDispatcherFactory df = serviceDescriptor
					.createDispatcherFactory();
			if (df != null) {
				this.setDispatcherFactory(df);
			} else if (logger.isDebugEnabled()) {
				logger.debug("Transport '" + getProtocol()
						+ "' will not support request endpoints: ");
			}

			if (logger.isDebugEnabled()) {
				logger.debug("Loading RequesterFactory for connector: "
						+ getName() + " (" + getClass().getName() + ")");
			}

			sessionHandler = serviceDescriptor.createSessionHandler();

		} catch (Exception e) {
			throw new InitialisationException(e, this);
		}
	}

	protected ITransportServiceDescriptor getServiceDescriptor() {
		if (serviceDescriptor == null) {
			throw new IllegalStateException(
					"This connector has not yet been initialised: " + name);
		}
		return serviceDescriptor;
	}

	protected IMessageReceiver createReceiver(IService service,
			IUuahEndpoint endpoint) throws Exception {
		return getServiceDescriptor().createMessageReceiver(this, service,
				endpoint);
	}

	public IMessageAdapter getMessageAdapter(Object message)
			throws MessagingException {
		try {
			return serviceDescriptor.createMessageAdapter(message);
		} catch (TransportServiceException e) {
			throw new MessagingException(ServerMessages
					.failedToCreate("Message Adapter"), message, e);
		}
	}

	public IUuahContext getUuahContext() {
		return uuahContext;
	}

	public void setUuahContext(IUuahContext context) {
		this.uuahContext = context;
		updateCachedNotificationHandler();
	}

	public String toString() {
		final StringBuffer sb = new StringBuffer(120);
		sb.append(ClassUtils.getSimpleName(this.getClass()));
		sb.append("{this=").append(
				Integer.toHexString(System.identityHashCode(this)));
		sb.append(", started=").append(started);
		sb.append(", initialised=").append(initialised);
		sb.append(", name='").append(name).append('\'');
		sb.append(", disposed=").append(disposed);
		sb.append(", connected=").append(connected);
		sb.append(", supportedProtocols=").append(supportedProtocols);
		sb.append('}');
		return sb.toString();
	}

	public Semaphore getConnectedSemaphore() {
		return connectedSemaphore;
	}

	public RetryPolicyTemplate getRetryPolicyTemplate() {
		return retryPolicyTemplate;
	}

	public void setRetryPolicyTemplate(RetryPolicyTemplate retryPolicyTemplate) {
		this.retryPolicyTemplate = retryPolicyTemplate;
	}

	protected void updateCachedNotificationHandler() {
		if (null != uuahContext) {
			notificationHandler = uuahContext.getNotificationManager();
		}
	}
}
